package com.clown.orderPayDetect

import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

// 定义到账事件样例类
case class ReceiptEvent(txId: String, payChannel: String, timestamp: Long)

object TxMatch {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    // 读取订单事件数据
    val orderEventStream = env.readTextFile(getClass.getResource("/OrderLog.csv").getPath)
      .map(var2 => {
        val arr = var2.split(",")
        OrderEvent(arr(0).toLong, arr(1), arr(2), arr(3).toLong)
      })
      .assignAscendingTimestamps(_.timestamp * 1000L)
      .filter(var2 => var2.eventType.equals("pay") && var2.txId != null)
      .keyBy(_.txId)

    // 读取到账事件数据
    val receiptEventStream = env.readTextFile(getClass.getResource("/ReceiptLog.csv").getPath)
      .map(var2 => {
        val arr = var2.split(",")
        ReceiptEvent(arr(0), arr(1), arr(2).toLong)
      })
      .assignAscendingTimestamps(_.timestamp * 1000L)
      .keyBy(_.txId)

    // 侧输出流标签
    val unMatchedPayEventOutputTag = new OutputTag[OrderEvent]("unMatchedPayEventOutputTag")
    val unMatchedReceiptEventOutputTag = new OutputTag[ReceiptEvent]("unMatchedReceiptEventOutputTag")

    // 合并两条流，进行处理
    val resultStream = orderEventStream.connect(receiptEventStream)
      .process(new TxPayMatchResult(unMatchedPayEventOutputTag, unMatchedReceiptEventOutputTag))

    resultStream.print("matched")
    resultStream.getSideOutput(unMatchedPayEventOutputTag).print("unMatched pays")
    resultStream.getSideOutput(unMatchedReceiptEventOutputTag).print("unMatched receipts")

    env.execute("tx match job")
  }
}

class TxPayMatchResult(payTag: OutputTag[OrderEvent], receiptTag: OutputTag[ReceiptEvent]) extends KeyedCoProcessFunction[Long, OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)] {
  // 定义状态，用于保存当前交易对应的订单支付事件和到账事件
  lazy val payEventState: ValueState[OrderEvent] = getRuntimeContext.getState(new ValueStateDescriptor[OrderEvent]("pay", classOf[OrderEvent]))
  lazy val receiptEventState: ValueState[ReceiptEvent] = getRuntimeContext.getState(new ValueStateDescriptor[ReceiptEvent]("receipt", classOf[ReceiptEvent]))


  override def processElement1(value: OrderEvent, ctx: KeyedCoProcessFunction[Long, OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]#Context, out: Collector[(OrderEvent, ReceiptEvent)]): Unit = {
    // 订单支付来了，判断之前是否有到账事件
    val receipt = receiptEventState.value()
    if (receipt != null) {
      // 如果已经有receipt，正常输出匹配，清空状态
      out.collect((value, receipt))
      receiptEventState.clear()
      payEventState.clear()
    } else {
      // 如果没来，注册定时器开始等待5秒
      ctx.timerService().registerEventTimeTimer(value.timestamp * 1000L + 5000L)
      // 更新状态
      payEventState.update(value)
    }
  }

  override def processElement2(value: ReceiptEvent, ctx: KeyedCoProcessFunction[Long, OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]#Context, out: Collector[(OrderEvent, ReceiptEvent)]): Unit = {
    // 到账支付来了，判断之前是否有订单事件
    val order = payEventState.value()
    if (order != null) {
      // 如果已经有receipt，正常输出匹配，清空状态
      out.collect((order, value))
      receiptEventState.clear()
      payEventState.clear()
    } else {
      // 如果没来，注册定时器开始等待5秒
      ctx.timerService().registerEventTimeTimer(value.timestamp * 1000L + 3000L)
      // 更新状态
      receiptEventState.update(value)
    }
  }

  override def onTimer(timestamp: Long, ctx: KeyedCoProcessFunction[Long, OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]#OnTimerContext, out: Collector[(OrderEvent, ReceiptEvent)]): Unit = {
    // 定时器触发，判断状态中哪个还存在，就代表另一个没来
    if (payEventState.value() != null) {
      ctx.output(payTag, payEventState.value())
    }
    if (receiptEventState.value() != null) {
      ctx.output(receiptTag, receiptEventState.value())
    }
    // 清空状态
    receiptEventState.clear()
    payEventState.clear()
  }
}
